package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo4DataFreamApi {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .appName("api")
      .config("spark.sql.shuffle.partitions", 1) // 设sql shuffle之后分区数, 默认是200
      .master("local")
      .getOrCreate()

    import spark.implicits._


    val df: DataFrame = spark.read.json("spark/data/students.json")


    //对多次使用的DF进行缓存
    df.cache()


    /**
      * show: 相当于一个action算子
      *
      * numRows： show 的行数
      * truncate: 是否完全显示数据
      *
      */

    df.show()
    df.show(100)
    df.show(false)


    /**
      *
      * select : 选择, 返回一个新的DF ,相当于map算子， 转换算子
      *
      */

    //指定列名，只能简单获取数据
    df.select("name", "age", "clazz").show()

    //传入一个列的对象，可以对列进行处理
    //as 取别名
    df.select($"name" as "newName", $"age" / 2 as "age").show()


    /**
      *
      * filter and where
      *
      */

    //传入一个字符串的表达式
    df.where("age > 22").show()

    //传入一个列的表达式
    df.where($"age" > 22).show()


    /**
      *
      * group: 分组之后必须接聚合函数
      *
      */

    df.groupBy("gender").count().show()


    //到spark sql 函数
    import org.apache.spark.sql.functions._

    //每个班级最大的年龄
    df.groupBy($"clazz").agg(max("age")).show()

    //班级人数
    df.groupBy($"clazz").agg(count("clazz") as "num").show()

    //统计班级平均年龄
    df.groupBy($"clazz").agg(mean($"age") as "avgage").show()
    df.groupBy($"clazz").agg(sum($"age") / count("clazz") as "avgAge").show()


    /**
      *
      * join
      *
      */

    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .schema("stuId STRING, cID STRING, sco INT")
      .load("spark/data/score.txt")


    //指定关联条件，指定关联方式
    val joinDF: DataFrame = df.join(scoreDF, $"id" === $"stuId", "inner")


    joinDF.groupBy($"id").agg(sum($"sco") as "sumScore").show()

    /**
      *
      * sql执行顺序
      *
      * select id,sum(sco) from stu join sco on id= id where age=23 group by id
      *
      *
      * from --> join --> on --> where ---> group by ---> having ---> select  --> order by ---> limit
      *
      *
      */

    /**
      *
      *
      *
      * 写sql需要注册临时视图
      *
      */

    //全局临时视图
    df.createGlobalTempView("a")
    //全局临时视图，
    df.createOrReplaceGlobalTempView("b")

    //创建局部临时视图
    df.createTempView("c")

    //创建局部临时视图   如果临时视图已存在就替换
    df.createOrReplaceTempView("student")


    spark.sql(
      """
        |
        |select clazz,count(*) as num from  student
        |group by clazz
        |
        |
      """.stripMargin).show()


    /**
      *
      * save
      *
      */

    val clazzNUm: DataFrame = df.groupBy("clazz").agg(count("clazz") as "num")


    clazzNUm
      .write
      .mode(SaveMode.Overwrite) //覆盖
      .csv("spark/data/clazzNum")


  }

}
